package drds.data_migrate.job;


import drds.common.Author;
import drds.data_migrate.Authors;
import drds.data_migrate.job.read_and_write.ReadJob;
import drds.data_migrate.job.read_and_write.WriteJob;
import drds.data_migrate.vo.JobConfig;
import drds.data_migrate.vo.TaskConfig;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

//import drds.data_migrate.job.wrapper.WriteJobWrapper;

//import drds.data_migrate.job.wrapper.ReadJobWrapper2;


/**
 * job实例运行在jobContainer容器中，它是所有任务的master,负责初始化,拆分,调度,运行,回收,监控和汇报,但它并不做实际的数据同步操作
 */
@Author(name = Authors.LI_YANG)
@Slf4j
public class JobContainer
{
    @Setter
    @Getter
    private JobConfig jobConfig;

    /**
     * reader和writer jobContainer的实例
     */
    private ReadJob readJob;

    private WriteJob writeJob;


    private int needChannelNumber;

    private int totalStage = 1;


    public JobContainer(JobConfig jobConfig)
    {
        this.jobConfig = jobConfig;
    }

    /**
     * jobContainer主要负责的工作全部在start()里面，包括init、prepare、doBigIntegerSplit、scheduler、
     * post以及destroy和statistics
     */

    public void run()
    {

        boolean hasException = false;
        try
        {


            this.init();

            this.prepare();

            this.totalStage = this.split();

            this.schedule();

            this.post();

            this.postHandle();


        } catch (Throwable e)
        {


            hasException = true;


            throw new RuntimeException(e);
        } finally
        {
            this.destroy();

            if (!hasException)
            {

            }
        }
    }


    private void adjustChannelNumber()
    {
        int needChannelNumberByByte = Integer.MAX_VALUE;
        int needChannelNumberByRecord = Integer.MAX_VALUE;


        // 取较小值
        this.needChannelNumber = needChannelNumberByByte < needChannelNumberByRecord ? needChannelNumberByByte : needChannelNumberByRecord;
        // 如果从byte或record上设置了needChannelNumber则退出
        if (this.needChannelNumber < Integer.MAX_VALUE)
        {
            return;
        }

        boolean isChannelLimit = false;//(this.taskConfig.getInt( Constants.job_setting_speed_channel, 0) > 0);
        if (isChannelLimit)
        {
            this.needChannelNumber = 0;//this.taskConfig.getInt( Constants.job_setting_speed_channel);

            log.info("job set RecordQueue-Number to " + this.needChannelNumber
                    + " channels.");

            return;
        }

        throw new RuntimeException("Job运行速度必须设置");
    }


    /**
     * reader和writer的初始化
     */
    private void init() throws SQLException
    {

        //必须先Reader ，后Writer
        this.readJob = new ReadJob(this.jobConfig);
        this.writeJob = new WriteJob(jobConfig);
        //
        this.readJob.init();
        this.writeJob.init();
    }

    private void prepare() throws SQLException
    {
        this.readJob.preCheck();
        this.writeJob.preCheck();
        //
        this.writeJob.prepare();
    }


    /**
     * 执行reader和writer最细粒度的切分，需要注意的是，writer的切分结果要参照reader的切分结果，
     * 达到切分后数目相等，才能满足1：1的通道模型，所以这里可以将reader和writer的配置整合到一起，
     * 然后，为避免顺序给读写端带来长尾影响，将整合的结果shuffler掉
     */
    private int split() throws SQLException
    {
        this.adjustChannelNumber();

        if (this.needChannelNumber <= 0)
        {
            this.needChannelNumber = 1;
        }

        List<TaskConfig> readerTaskTaskConfigs = this.doReaderSplit(this.needChannelNumber);
        int taskNumber = readerTaskTaskConfigs.size();
        List<TaskConfig> writerTaskTaskConfigs = this.doWriterSplit(taskNumber);


        /**
         * 输入是reader和writer的parameter list，输出是content下面元素的list
         */
        List<TaskConfig> contentTaskConfig = mergeReaderAndWriterTaskConfigs(
                readerTaskTaskConfigs, writerTaskTaskConfigs);


        //this.taskConfig.set(Constants.taskConfigList, contentTaskConfig);

        return contentTaskConfig.size();
    }

    // TODO: 如果源头就是空数据
    private List<TaskConfig> doReaderSplit(int adviceNumber) throws SQLException
    {

        List<TaskConfig> readerSlicesTaskConfigs = this.readJob.split(adviceNumber);
        if (readerSlicesTaskConfigs == null || readerSlicesTaskConfigs.size() <= 0)
        {
            throw new RuntimeException("reader切分的task数目不能小于等于0");
        }


        return readerSlicesTaskConfigs;
    }

    private List<TaskConfig> doWriterSplit(int readerTaskNumber)
    {


        List<TaskConfig> writerSlicesTaskConfigs = this.writeJob
                .split(readerTaskNumber);
        if (writerSlicesTaskConfigs == null || writerSlicesTaskConfigs.size() <= 0)
        {
            throw new RuntimeException("writer切分的task不能小于等于0");
        }


        return writerSlicesTaskConfigs;
    }

    /**
     * 按顺序整合reader和writer的配置，这里的顺序不能乱！ 输入是reader、writer级别的配置，输出是一个完整task的配置
     */
    private List<TaskConfig> mergeReaderAndWriterTaskConfigs(
            List<TaskConfig> readerTasksTaskConfigs,
            List<TaskConfig> writerTasksTaskConfigs)
    {
        if (readerTasksTaskConfigs.size() != writerTasksTaskConfigs.size())
        {
            throw new RuntimeException("reader切分的task数目[%d]不等于writer切分的task数目[%d]." +
                    readerTasksTaskConfigs.size() + writerTasksTaskConfigs.size()
            );
        }

        List<TaskConfig> contentTaskConfigs = new ArrayList<TaskConfig>();
        for (int i = 0; i < readerTasksTaskConfigs.size(); i++)
        {
            TaskConfig taskConfig = new TaskConfig();

            taskConfig.setJobReaderParameter(readerTasksTaskConfigs.get(i));

            taskConfig.setJobWriterParameter(
                    writerTasksTaskConfigs.get(i));


            taskConfig.setTaskId(i);
            contentTaskConfigs.add(taskConfig);
        }

        return contentTaskConfigs;
    }

    /**
     * schedule首先完成的工作是把上一步reader和writer split的结果整合到具体taskGroupContainer中,
     * 同时不同的执行模式调用不同的调度策略，将所有任务调度起来
     */
    private void schedule()
    {
        /**
         * 这里的全局speed和每个channel的速度设置为B/s
         */

        int taskNumber = 0;//this.taskConfig.getList(Constants.taskConfigList).size();
        int channelsPerTaskGroup = 5;//this.taskConfig.getInt(Constants.readWriteTaskNumberOfThisTaskGroup, 5);
        //
        this.needChannelNumber = Math.min(this.needChannelNumber, taskNumber);
        //PerfTrace.getInstance().setChannelNumber(needChannelNumber);

        /**
         * 通过获取配置信息得到每个taskGroup需要运行哪些tasks任务
         */

        List<TaskGroup> taskGroupTaskConfigList = TaskAssigns.resourceLoadBalanceAssign(null, this.needChannelNumber, channelsPerTaskGroup);

        Scheduler scheduler;
        try
        {

            scheduler = initScheduler();


            scheduler.schedule(taskGroupTaskConfigList);


        } catch (Exception e)
        {


            throw new RuntimeException(e);
        }


    }

    private Scheduler initScheduler()
    {

        return new Scheduler();
    }

    private void post() throws SQLException
    {
        this.writeJob.post();
        this.readJob.post();
    }


    private void postHandle() throws SQLException
    {


    }

    private void destroy()
    {
        if (this.writeJob != null)
        {
            this.writeJob.destroy();
            this.writeJob = null;
        }
        if (this.readJob != null)
        {
            this.readJob.destroy();
            this.readJob = null;
        }
    }


}
